package com.chukun.flink.table.api.table;

import com.chukun.flink.table.bean.ClickBean;
import com.chukun.flink.table.source.PrepareData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Timestamp;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author chukun
 * @version 1.0.0
 * @description table api基本使用 <a href="https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/tableapi/">详细算子参考</a>
 * @createTime 2022年06月03日 11:52:00
 */
public class TableApiTemplate {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<ClickBean> dataStream = env.fromCollection(PrepareData.getClicksData());

        Table table = tableEnv.fromDataStream(dataStream, Schema.newBuilder()
                .column("id", DataTypes.INT())
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("time", DataTypes.TIMESTAMP().bridgedTo(Timestamp.class))
                .build());

        tableEnv.createTemporaryView("Clicks", table);

        Table orders = tableEnv.from("Clicks");

        // table api算子的基本使用，详细参考: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/tableapi/
        Table result = orders.filter($("user").in("张三", "李四"))
                .select($("id"),$("user"),$("time"));

        DataStream<Row> rowDataStream = tableEnv.toDataStream(result, Row.class);
        rowDataStream.print("result");
        env.execute();
    }
}
